package com.tiantian.kafka.learn_08_transactional;

import org.apache.kafka.clients.consumer.*;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.apache.kafka.common.serialization.StringSerializer;

import java.time.Duration;
import java.util.*;
import java.util.concurrent.Future;

/**
 * @ClassName TransactionProgram
 * @Description TODO
 * @Author tant
 * @Date 2025/3/5 14:01
 */
@SuppressWarnings("all")
public class TransactionProgram {


    /**
     * 消费者从ods_user主题拉去数据，进行预处理，并使用生产者写入到dwd_user主题中
     *
     * @param args
     */
    public static void main(String[] args) {
        // 1. 调用之前实现的方法，创建消费者、生产者对象
        KafkaConsumer<String, String> consumer = createConsumer();
        KafkaProducer<String, String> producer = createProducer();

        // 2. 生产者调用initTransactions初始化事务
        producer.initTransactions();

        // 3. 编写一个while死循环，在while循环中不断拉取数据，进行处理后，再写入到指定的topic
        while (true) {
            try {
                // (1)  生产者开启事务
                producer.beginTransaction();

                // 这个Map保存了topic对应的partition的偏移量
                Map<TopicPartition, OffsetAndMetadata> offsetMap = new HashMap<>();

                // 从topic中拉取一批的数据
                // (2)  消费者拉取消息
                ConsumerRecords<String, String> concumserRecordArray = consumer.poll(Duration.ofSeconds(5));
                // (3)  遍历拉取到的消息，并进行预处理
                for (ConsumerRecord<String, String> cr : concumserRecordArray) {
                    System.out.printf("消费者-topic: %s partition: %d offset: %d key: %s msg: %s%n", cr.topic(), cr.partition(), cr.offset(), cr.key(), cr.value());

                    //模拟异常
                    int nex = new Random().nextInt(5);
                    if (nex == 0) {
                        int i = 1 / 0;
                    }

                    // offset + 1：offset是当前消费的记录（消息）对应在partition中的offset，而我们希望下一次能继续从下一个消息消息
                    // 必须要+1，从能消费下一条消息
                    offsetMap.put(new TopicPartition(cr.topic(), cr.partition()), new OffsetAndMetadata(cr.offset() + 1));


                    // 消费的是ods_user中的数据,将1转换为男，0转换为女
                    String msg = preHandleMsg(cr.value());


                    // (4)  生产消息到dwd_user topic中
                    ProducerRecord<String, String> dwdMsg = new ProducerRecord<>("dwd_user", null, msg);
                    // 发送消息::如果使用了事务，不要使用异步发送
                    Future<RecordMetadata> future = producer.send(dwdMsg);
                    try {
                        RecordMetadata metadata = future.get();

                        System.out.printf("消息发送成功: topic: %s partition: %d offset: %d key: %s value: %s%n", metadata.topic(), metadata.partition(), metadata.offset(), dwdMsg.key(), dwdMsg.value());
                    } catch (Exception e) {
                        e.printStackTrace();
                        producer.abortTransaction();
                    }
                }

                //  (5)  保存消费的offset
                producer.sendOffsetsToTransaction(offsetMap, "ods_user");

                // (6)  提交事务
                producer.commitTransaction();
            } catch (Exception e) {
                e.printStackTrace();
                // (7)  捕获异常，如果出现异常，则取消事务
                producer.abortTransaction();
            }
        }
    }


    /**
     * 预处理消息~
     *
     * @param msg
     */
    private static String preHandleMsg(String msg) {
        String[] fieldArray = msg.split(",");
        // 将字段进行替换
        if (fieldArray != null && fieldArray.length > 2) {
            String sexField = fieldArray[1];
            if (sexField.equals("1")) {
                fieldArray[1] = "男";
            } else if (sexField.equals("0")) {
                fieldArray[1] = "女";
            }
        }
        //  重新拼接字段
        return fieldArray[0] + "," + fieldArray[1] + "," + fieldArray[2];
    }

    /**
     * 创建一个消费者来消费ods_user中的数据
     * - 配置事务的隔离级别
     * - 关闭自动提交，一会我们需要手动来提交offset，通过事务来维护offset
     *
     * @return
     */
    private static KafkaConsumer<String, String> createConsumer() {
        // 1. 配置消费者的属性（添加对事务的支持）
        Properties props = new Properties();
        props.setProperty(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.217.132:9092");
        //配置key和value的序列化方式
        props.setProperty(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        props.setProperty(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        // 配置事务的隔离级别
        props.setProperty(ConsumerConfig.ISOLATION_LEVEL_CONFIG, "read_committed");
        // 关闭自动提交，一会我们需要手动来提交offset，通过事务来维护offset
        props.setProperty(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false");
        // 设置消费组id
        props.setProperty(ConsumerConfig.GROUP_ID_CONFIG, "ods_user");
        // 2. 构建消费者对象
        KafkaConsumer<String, String> kafkaConsumer = new KafkaConsumer<>(props);
        // 3. 订阅一个topic
        kafkaConsumer.subscribe(Arrays.asList("ods_user"));
        return kafkaConsumer;

    }

    /**
     * 编写createProducer方法，用来创建一个带有事务配置的生产者
     * - 开启事务必须要配置事务的ID
     *
     * @return
     */
    private static KafkaProducer<String, String> createProducer() {
        // 1. 配置生产者带有事务配置的属性
        Properties props = new Properties();
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.217.132:9092");
        //配置key和value的序列化方式
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        // 开启事务必须要配置事务的ID
        props.put(ProducerConfig.TRANSACTIONAL_ID_CONFIG, "dwd_user");
        // 2. 构建生产者
        KafkaProducer<String, String> kafkaProducer = new KafkaProducer<>(props);
        return kafkaProducer;
    }
}


    